package com.lxj.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.lxj.utils.DateFormatUtil;
import com.lxj.utils.MyKafkaUtil;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;


//流量域未经加工的事务事实表
public class BaseLogApp {
    /**
     * 主方法，负责构建并执行Flink数据处理流程
     * 主要功能：
     * 1. 从Kafka读取原始日志数据
     * 2. 数据清洗和格式转换
     * 3. 新老用户状态标记修复
     * 4. 数据分流处理
     * 5. 将处理后的数据写入不同的Kafka主题
     *
     * .
     * BaseLogApp.java （基础日志处理）
     *
     * - 从"topic_log"主题读取原始日志
     * - 进行数据清洗、新老访客标记修复
     * - 将处理后的数据分流写入"dwd_traffic_page_log"等主题
     * - 这是整个DWD层的基础，必须首先运行
     *
     * @param args 命令行参数
     * @throws Exception 执行过程中可能抛出的异常
     */

    public static void main(String[] args) throws Exception {

        // TODO 1. 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        // TODO 2. 启用状态后端

        // TODO 3. 从 Kafka 读取主流数据
        String topic = "topic_log";
        String groupId = "base_log_consumer";

        // 创建Kafka数据源
        DataStreamSource<String> source = env
                .addSource(MyKafkaUtil
                        .getFlinkKafkaConsumer(topic, groupId));
//        source.print();

        // 定义脏数据侧输出流标签
        OutputTag<String> dirtyStreamTag = new OutputTag<String>("dirtyStream") {
        };

        /**
         * 数据清洗处理函数
         * 主要功能：
         * 1. 尝试解析JSON格式数据
         * 2. 将格式正确的数据发送到主流
         * 3. 将格式错误的数据作为脏数据处理
         */
        SingleOutputStreamOperator<String> cleanedStream = source.process(
                new ProcessFunction<String, String>() {
                    @Override
                    public void processElement(String jsonStr, Context ctx, Collector<String> out) throws Exception {
                        try {
                            // 尝试解析JSON字符串，验证格式是否正确
                            out.collect(jsonStr);
                        } catch (Exception e) {
                            // 捕获异常，处理脏数据
                            System.out.println(jsonStr);
                        }
                    }
                }
        );


        // 将清洗后的JSON字符串转换为JSONObject对象
        SingleOutputStreamOperator<JSONObject> mappedStream = cleanedStream.map(JSON::parseObject);

        // 按照mid(设备ID)对数据进行分组
        KeyedStream<JSONObject, String> keyedStream = mappedStream
                .keyBy(r -> r.getJSONObject("common").getString("mid"));//mid相同的给共享

        /**
         * 新老访客状态标记修复处理函数
         * 主要功能：
         * 1. 维护每个mid的首次访问日期状态
         * 2. 修正is_new标记错误的情况
         */
        SingleOutputStreamOperator<JSONObject> fixedStream = keyedStream.process(
                new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    // 定义状态，保存每个mid的首次访问日期
                    ValueState<String> firstViewDtState;

                    @Override
                    public void open(Configuration param) throws Exception {
                        super.open(param);
                        // 初始化状态
                        firstViewDtState = getRuntimeContext().getState(new ValueStateDescriptor<String>("firstViewDtState",
                                String.class));
                    }
                    @Override
                    public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                        String isNew = jsonObj.getJSONObject("common").getString("is_new");
                        String firstViewDt = firstViewDtState.value(); // 获取状态中保存的首次访问日期
                        Long ts = jsonObj.getLong("ts"); // 获取当前记录的时间戳
                        String dt = DateFormatUtil.toDate(ts); // 转换为日期格式

                        if ("1".equals(isNew)) {
                            // 处理标记为新用户的情况
                            if (firstViewDt == null) {
                                // 首次访问，更新状态
                                firstViewDtState.update(dt);
                            } else if (!firstViewDt.equals(dt)) {
                                // 非首次访问，修正is_new标记
                                isNew = "0";
                                jsonObj.getJSONObject("common").put("is_new", isNew);
                            }
                        } else {
                            // 处理标记为老用户的情况
                            if (firstViewDt == null) {
                                // 将首次访问日期置为昨日
                                String yesterday = DateFormatUtil.toDate(ts - 1000 * 60 * 60 * 24);
                                firstViewDtState.update(yesterday);
                            }
                        }
                        out.collect(jsonObj);
                    }
                }

        );



        // 定义各种日志类型的侧输出流标签
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> errorTag = new OutputTag<String>("errorTag") {
        };

        /**
         * 数据分流处理函数
         * 主要功能：
         * 1. 将错误数据分流到errorTag
         * 2. 将启动数据分流到startTag
         * 3. 将曝光数据分流到displayTag
         * 4. 将动作数据分流到actionTag
         * 5. 将页面数据保留在主流
         */

        SingleOutputStreamOperator<String> separatedStream = fixedStream.process(
                new ProcessFunction<JSONObject, String>() {
                    @Override
                    public void processElement(JSONObject jsonObj, Context context, Collector<String> out) throws Exception {
                        // 处理错误数据
                        JSONObject error = jsonObj.getJSONObject("err");
                        if (error != null) {
                            context.output(errorTag, jsonObj.toJSONString());
                        }
                        jsonObj.remove("err");

                        // 处理启动数据
                        JSONObject start = jsonObj.getJSONObject("start");
                        if (start != null) {
                            context.output(startTag, jsonObj.toJSONString());
                        } else {
                            // 处理页面数据
                            JSONObject page = jsonObj.getJSONObject("page");
                            JSONObject common = jsonObj.getJSONObject("common");
                            Long ts = jsonObj.getLong("ts");

                            // 处理曝光数据
                            JSONArray displays = jsonObj.getJSONArray("displays");
                            if (displays != null) {
                                for (int i = 0; i < displays.size(); i++) {
                                    JSONObject display = displays.getJSONObject(i);
                                    JSONObject displayObj = new JSONObject();
                                    displayObj.put("display", display);
                                    displayObj.put("common", common);
                                    displayObj.put("page", page);
                                    displayObj.put("ts", ts);
                                    context.output(displayTag, displayObj.toJSONString());
                                }
                            }

                            // 处理动作数据
                            JSONArray actions = jsonObj.getJSONArray("actions");
                            if (actions != null) {
                                for (int i = 0; i < actions.size(); i++) {
                                    JSONObject action = actions.getJSONObject(i);
                                    JSONObject actionObj = new JSONObject();
                                    actionObj.put("action", action);
                                    actionObj.put("common", common);
                                    actionObj.put("page", page);
                                    actionObj.put("ts", ts);
                                    context.output(actionTag, actionObj.toJSONString());
                                }
                            }

                            // 清理不需要的字段后输出到主流
                            jsonObj.remove("displays");
                            jsonObj.remove("actions");
                            out.collect(jsonObj.toJSONString());
                        }
                    }
                }
        );

        // 从侧输出流获取各种类型的数据流
        DataStream<String> startDS = separatedStream.getSideOutput(startTag);
        DataStream<String> displayDS = separatedStream.getSideOutput(displayTag);
        DataStream<String> actionDS = separatedStream.getSideOutput(actionTag);
        DataStream<String> errorDS = separatedStream.getSideOutput(errorTag);

        // 定义Kafka输出主题名称
        String page_topic = "dwd_traffic_page_log";
        String start_topic = "dwd_traffic_start_log";
        String display_topic = "dwd_traffic_display_log";
        String action_topic = "dwd_traffic_action_log";
        String error_topic = "dwd_traffic_error_log";

        // 将各数据流写入对应的Kafka主题
        separatedStream.print("page>>>>>>>>>>>>>>>>>>>");
        separatedStream.addSink(MyKafkaUtil.getFlinkKafkaProducer(page_topic));
        startDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(start_topic));
        displayDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(display_topic));
        actionDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(action_topic));
        errorDS.addSink(MyKafkaUtil.getFlinkKafkaProducer(error_topic));

        // 执行Flink作业
        env.execute();



    }
}
